package com.streamsets.datacollector.pileline.stage;

import com.streamsets.datacollector.model.etl.ColumnValueReplaceConfigJson;
import com.streamsets.datacollector.model.etl.bean.ColumnToReplaceBean;
import com.streamsets.datacollector.restapi.bean.ConfigConfigurationJson;
import com.streamsets.datacollector.restapi.bean.StageConfigurationJson;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 *  初始化 列值替换(某一列的值都替换为某个值)的Stage的配置
 *  stage的完整描述示例如：
 "configuration" : [ {
 "name" : "conf.rules",
 "value" : [ {
 "setToNull" : false,
 "fields" : "/customer_id",
 "replacement" : "666"
 } ]
 },....]
 *
 * @author wengzhonghui
 * @date 2020/9/16 10:32
 */
public class InitColumnValReplaceConfig {

    /*
     * stage instanceName名称
     */
    private static final String STAGE_NAME = "g3_01";

    /*
     * 名称
     */
    private static final String CONFIG_NAME_REPLACE = "conf.rules";


    public static boolean init(StageConfigurationJson stage, ColumnValueReplaceConfigJson configDto){
        if(STAGE_NAME.equalsIgnoreCase(stage.getInstanceName())){
            if (configDto != null) {
                List<ConfigConfigurationJson> configs = intConfig(stage.getConfiguration(), configDto);
                stage.setConfiguration(configs);
            } else {
                List<ConfigConfigurationJson> configs = stage.getConfiguration().stream().filter(e -> "conf.rules".equals(e.getName())).collect(Collectors.toList());
                stage.getConfiguration().removeAll(configs);
            }
        }
        return true;
    }

    public static List<ConfigConfigurationJson> intConfig(List<ConfigConfigurationJson> oldConfigs, ColumnValueReplaceConfigJson configDto){
        List<ConfigConfigurationJson> newConfigs = new ArrayList<>();
        for(ConfigConfigurationJson config :oldConfigs ){
            config = initColumnReplacerConfigs(config, configDto.getColumnToReplaceConfigs());
            newConfigs.add(config);
        }
        return newConfigs;
    }


    /*
    * 把 列值替换为的配置
    * 格式示例如：
    * [ {
         "setToNull" : false,
         "fields" : "/customer_id",
         "replacement" : "666"
       } ]
    * @param [config 原来的配置, columnNames:需要替换为null的列名列表]
    * @return 返回新生成的配置
    */
    private static ConfigConfigurationJson initColumnReplacerConfigs(ConfigConfigurationJson config, List<ColumnToReplaceBean> columnToReplaceConfigs){
        if(CONFIG_NAME_REPLACE.equalsIgnoreCase(config.getName())){
            List configValList = new ArrayList();
            if(columnToReplaceConfigs != null && columnToReplaceConfigs.size() > 0){
                for(ColumnToReplaceBean valConfig : columnToReplaceConfigs){
                    if(valConfig == null  ||  StringUtils.isEmpty(valConfig.getFields())){
                        continue;
                    }
                    Map<String,Object> data = new HashMap();
                    data.put("fields", "/" + valConfig.getFields());
                    String setToNull = valConfig.getSetToNull();
                    if(StringUtils.isNotEmpty(setToNull) && "1".equalsIgnoreCase(setToNull)){
                        data.put("setToNull", true);
                    }else{
                        data.put("setToNull", false);
                    }
                    String newValue = StringUtils.isEmpty(valConfig.getReplacement()) ?  "": valConfig.getReplacement();
                    data.put("replacement", newValue);
                    configValList.add(data);
                }
            }
            config = new ConfigConfigurationJson(CONFIG_NAME_REPLACE,configValList);
        }
        return config;
    }


}
